package com.suzus.four;


import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.suzus.utils.RabbitMqUtils;

import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

public class ConfirmMessage {

    public static final int MESSAGE_COUNT = 1000;

    public static void main(String[] args) throws Exception {
        ConfirmMessage.publicMessageAsync();
    }

    public static void publicMessageIndividual() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        String string = UUID.randomUUID().toString();
        channel.queueDeclare(string,true,false,false,null);

        channel.confirmSelect();

        long begin = System.currentTimeMillis();

        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("",string,null,message.getBytes());
            boolean flag = channel.waitForConfirms();
            if(flag){
                System.out.println("success");
            } else {
                System.out.println("fail");
            }
        }

        long end = System.currentTimeMillis();

        System.out.println(end-begin);

    }

    public static void publicMessageBatch() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        String string = UUID.randomUUID().toString();
        channel.queueDeclare(string,true,false,false,null);

        channel.confirmSelect();

        long begin = System.currentTimeMillis();

        int batchSize=1000;

        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("",string,null,message.getBytes());

            if(i%batchSize==0) {
                boolean flag = channel.waitForConfirms();
                if(flag){
                    System.out.println("success");
                } else {
                    System.out.println("fail");
                }
            }

        }

        long end = System.currentTimeMillis();

        System.out.println(end-begin);

    }

    public static void publicMessageAsync() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        String string = UUID.randomUUID().toString();
        channel.queueDeclare(string,true,false,false,null);

        channel.confirmSelect();

        ConcurrentSkipListMap<Long,String> outstandingConfirms = new ConcurrentSkipListMap<>();

        long begin = System.currentTimeMillis();

        ConfirmCallback var1 = (deliverTag,multiple) -> {
            if (multiple) {
                ConcurrentNavigableMap<Long,String>confirmed=outstandingConfirms.headMap(deliverTag);
                confirmed.clear();
            } else {
                outstandingConfirms.remove(deliverTag);
            }
            System.out.println("Confirm : "+deliverTag);
        };
        ConfirmCallback var2 = (deliverTag,multiple)-> {
            String s = outstandingConfirms.get(deliverTag);
            System.out.println("Not Confirm : "+s);
        };

        channel.addConfirmListener(var1, var2);

        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("",string,null,message.getBytes());

            outstandingConfirms.put(channel.getNextPublishSeqNo(),message);

        }

        long end = System.currentTimeMillis();

        System.out.println(end-begin);

    }

}
